package jdk10;

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.LongStream;


@Slf4j
public class FluxTest {

    @Test
    public void test() {
        Flux.fromIterable(LongStream.range(1, 500).boxed().collect(Collectors.toList()))
                .mergeWith(Flux.interval(Duration.ofMillis(100)))
                .doOnNext((l) -> log.info("this value :{}", l))
                .map(d -> d * 2)
                .take(3)
                .onErrorResume((e) -> {
                    log.error(e.getMessage(), e);
                    return new Publisher<Long>() {
                        @Override
                        public void subscribe(Subscriber<? super Long> subscriber) {
                            log.info(subscriber.toString());
                        }
                    };
                })
                .doAfterTerminate(() -> log.info("xx"))
                .subscribe(System.out::println);
        Flux.just(1, 2, 3, 4, 5, 6).subscribe(System.out::println);
        System.out.println(Flux.just(1, 2, 3, 4, 5, 6).getPrefetch());
    }

    @Test
    public void testBackpressure() {
        Flux.range(1, 6)
                .doOnRequest(n -> System.out.println("Request " + n + " values..."))
                .subscribe(new BaseSubscriber<Integer>() {
                    @Override
                    protected void hookOnSubscribe(Subscription subscription) {
                        System.out.println("Subscribed and make a request...");
                        request(1);
                    }

                    @Override
                    protected void hookOnNext(Integer value) {
                        try {
                            TimeUnit.SECONDS.sleep(1);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        System.out.println("Get value [" + value + "]");
                        request(1);
                    }
                });
    }

}
